import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.io.*;
import java.util.Properties;

public class Q1_KafkaProducer {
    public static void main(String[] args) throws InterruptedException {
        String brokers = "192.168.80.131:9092,192.168.80.132:9092,192.168.80.133:9092";
        String topic = "topicB";
        Properties prop = new Properties();

        prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
        prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);


        File f = new File("F:\\TrainingCourse\\P1 大数据\\大数据正式班第四阶段模块三 Spark下\\HomeWork4-3\\src\\main\\resources\\sample.log");
        try (BufferedReader reader = new BufferedReader(new FileReader(f));
             Producer<String, String> producer = new KafkaProducer<>(prop);) {
            String data = null;
            while ((data = reader.readLine()) != null) {
                producer.send(new ProducerRecord<>(topic, data, data));
                Thread.sleep(100);
                System.out.println("sent a line");
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
